// Copyright 2021 Matrix Origin
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package gc

import (
	"bytes"
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/matrixorigin/matrixone/pkg/common/moerr"
	"github.com/matrixorigin/matrixone/pkg/common/mpool"
	"github.com/matrixorigin/matrixone/pkg/container/batch"
	"github.com/matrixorigin/matrixone/pkg/container/types"
	"github.com/matrixorigin/matrixone/pkg/fileservice"
	"github.com/matrixorigin/matrixone/pkg/logutil"
	"github.com/matrixorigin/matrixone/pkg/objectio/ioutil"
	"github.com/matrixorigin/matrixone/pkg/util/fault"
	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/db/checkpoint"
	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logstore/wal"
	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logtail"
	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/testutils"
	"go.uber.org/zap"
)

type checkpointCleaner struct {
	ctx context.Context

	// TODO: remove `sid`
	sid string

	mp *mpool.MPool
	fs fileservice.FileService

	logDriver     wal.Store
	checkpointCli checkpoint.Runner
	deleter       *Deleter

	watermarks struct {
		// scanWaterMark is the watermark of the incremental checkpoint which has been
		// scanned by the cleaner. After the cleaner scans the checkpoint, it
		// records all the object-list found in the checkpoint into some GC-specific
		// files. The scanWaterMark is used to record the end of the checkpoint.
		// For example:
		// Incremental checkpoint: [t100, t200), [t200, t300), [t300, t400)
		// scanWaterMark: [t100, t200)
		// remainingObjects: windows: [t100, t200), [f1, f2, f3]
		// The cleaner will scan the checkpoint [t200, t300) next time. Then:
		// scanWaterMark: [t100, t200), [t200, t300)
		// remainingObjects: windows:
		// {[t100, t200), [f1, f2, f3]}, {[t200, t300), [f4, f5, f6]}
		scanWaterMark atomic.Pointer[checkpoint.CheckpointEntry]

		// gcWaterMark is the watermark of the global checkpoint which has been GC'ed.
		// when the cleaner GCs the global checkpoint, it scans all the object-list
		// of the global checkpoint and build a bloom filter for the object-list.
		// Then it scans all the remainingObjects generated by `incremental-checkpoint-scan`
		// and filters out the objects that are not in the bloom filter: canGC and cannotGC.
		// it records the cannotGC objects into some GC-specific files and do fine filter on
		// the canGC objects. The gcWaterMark is used to record the end of the global checkpoint
		// when the cleaner finishes this GC operation.
		gcWaterMark atomic.Pointer[checkpoint.CheckpointEntry]

		// checkpointGCWaterMark is the watermark after being merged. The checkpoint runner will
		// GC the entry in the checkpoint tree according to this watermark And the diskcleaner
		// will use this watermark as the start, and use `gcWaterMark` as the end to call `ICKPRange`,
		// to get the ickp to perform a merge operation
		checkpointGCWaterMark atomic.Pointer[types.TS]
	}

	options struct {
		gcEnabled           atomic.Bool
		checkEnabled        atomic.Bool
		gcCheckpointEnabled atomic.Bool
	}

	config struct {
		canGCCacheSize          int
		maxMergeCheckpointCount int
		estimateRows            int
		probility               float64
	}

	// checker is to check whether the checkpoint can be consumed
	checker struct {
		sync.RWMutex
		extras map[string]func(item any) bool
	}

	// backup protection mechanism
	backupProtection struct {
		sync.RWMutex
		// protectedTS is the timestamp that should be protected from GC
		protectedTS types.TS
		// lastUpdateTime is the last time the protection was updated
		lastUpdateTime time.Time
		// isActive indicates if backup protection is active
		isActive bool
	}

	mutation struct {
		sync.Mutex
		taskState struct {
			id        uint64
			name      string
			startTime time.Time
		}
		scanned      *GCWindow
		metaFiles    map[string]ioutil.TSRangeFile
		snapshotMeta *logtail.SnapshotMeta
		replayDone   bool
		// backupProtectionSnapshot is a snapshot of backup protection state at the start of GC
		// This ensures GC consistency: once GC starts, it uses this snapshot and ignores
		// any changes to backup protection during GC execution
		backupProtectionSnapshot struct {
			protectedTS types.TS
			isActive    bool
		}
	}

	// iscpTablesFunc is an optional function to provide the ISCP tables for testing.
	iscpTablesFunc func() (map[uint64]types.TS, error)
}

func WithCanGCCacheSize(
	size int,
) CheckpointCleanerOption {
	return func(e *checkpointCleaner) {
		e.config.canGCCacheSize = size
	}
}

func WithMaxMergeCheckpointCount(
	count int,
) CheckpointCleanerOption {
	return func(e *checkpointCleaner) {
		e.config.maxMergeCheckpointCount = count
	}
}

func WithGCProbility(
	probility float64,
) CheckpointCleanerOption {
	return func(e *checkpointCleaner) {
		e.config.probility = probility
	}
}

func WithEstimateRows(
	rows int,
) CheckpointCleanerOption {
	return func(e *checkpointCleaner) {
		e.config.estimateRows = rows
	}
}

// for ut
func WithGCCheckpointOption(enable bool) CheckpointCleanerOption {
	return func(e *checkpointCleaner) {
		e.options.gcCheckpointEnabled.Store(enable)
	}
}

func WithCheckOption(enable bool) CheckpointCleanerOption {
	return func(e *checkpointCleaner) {
		e.options.checkEnabled.Store(enable)
	}
}

type CheckpointCleanerOption func(*checkpointCleaner)

func NewCheckpointCleaner(
	ctx context.Context,
	sid string,
	fs fileservice.FileService,
	logDriver wal.Store,
	checkpointCli checkpoint.Runner,
	opts ...CheckpointCleanerOption,
) Cleaner {
	cleaner := &checkpointCleaner{
		ctx:           ctx,
		sid:           sid,
		fs:            fs,
		logDriver:     logDriver,
		checkpointCli: checkpointCli,
	}
	for _, opt := range opts {
		opt(cleaner)
	}
	cleaner.deleter = NewDeleter(fs)
	cleaner.options.gcEnabled.Store(true)
	cleaner.mp = common.CheckpointAllocator
	cleaner.checker.extras = make(map[string]func(item any) bool)
	cleaner.mutation.metaFiles = make(map[string]ioutil.TSRangeFile)
	cleaner.mutation.snapshotMeta = logtail.NewSnapshotMeta()
	cleaner.backupProtection.isActive = false
	return cleaner
}

func (c *checkpointCleaner) Stop() {
	c.mutation.Lock()
	defer c.mutation.Unlock()
	c.mutation.scanned = nil
	c.mutation.metaFiles = nil
	c.mutation.snapshotMeta = nil
}

func (c *checkpointCleaner) GetMPool() *mpool.MPool {
	return c.mp
}

func (c *checkpointCleaner) SetTid(tid uint64) {
	c.mutation.Lock()
	defer c.mutation.Unlock()
	c.mutation.snapshotMeta.SetTid(tid)
}

func (c *checkpointCleaner) EnableGC() {
	c.options.gcEnabled.Store(true)
}

func (c *checkpointCleaner) DisableGC() {
	c.options.gcEnabled.Store(false)
}

func (c *checkpointCleaner) GCEnabled() bool {
	return c.options.gcEnabled.Load()
}

func (c *checkpointCleaner) GCCheckpointEnabled() bool {
	return c.options.gcCheckpointEnabled.Load()
}

func (c *checkpointCleaner) EnableCheck() {
	c.options.checkEnabled.Store(true)
}
func (c *checkpointCleaner) DisableCheck() {
	c.options.checkEnabled.Store(false)
}

func (c *checkpointCleaner) CheckEnabled() bool {
	return c.options.checkEnabled.Load()
}

func (c *checkpointCleaner) StartMutationTask(name string) {
	c.mutation.Lock()
	c.mutation.taskState.id++
	c.mutation.taskState.name = fmt.Sprintf("%s-%d", name, c.mutation.taskState.id)
	c.mutation.taskState.startTime = time.Now()
}

func (c *checkpointCleaner) StopMutationTask() {
	logutil.Info(
		"GC-Task-Done",
		zap.String("task", c.TaskNameLocked()),
		zap.Duration("duration", time.Since(c.mutation.taskState.startTime)),
	)
	c.mutation.taskState.name = ""
	c.mutation.Unlock()
}

func (c *checkpointCleaner) TaskNameLocked() string {
	return c.mutation.taskState.name
}

func (c *checkpointCleaner) Replay(inputCtx context.Context) (err error) {
	now := time.Now()

	ctx, cancel := context.WithCancelCause(inputCtx)
	defer cancel(nil)
	go func() {
		select {
		case <-c.ctx.Done():
			cancel(context.Cause(c.ctx))
		case <-inputCtx.Done():
			cancel(context.Cause(inputCtx))
		case <-ctx.Done():
		}
	}()

	select {
	case <-ctx.Done():
		err = context.Cause(ctx)
		return
	default:
	}

	c.StartMutationTask("gc-replay")
	defer c.StopMutationTask()

	if c.mutation.replayDone {
		logutil.Warn(
			"GC-REPLAY-ALREADY-DONE",
			zap.String("task", c.TaskNameLocked()),
		)
		return
	}

	defer func() {
		if err == nil {
			c.mutation.replayDone = true
		} else {
			c.mutation.metaFiles = make(map[string]ioutil.TSRangeFile)
			c.mutation.snapshotMeta = logtail.NewSnapshotMeta()
			c.mutation.scanned = nil
		}
		logutil.Info(
			"GC-REPLAY-TRACE",
			zap.String("task", c.TaskNameLocked()),
			zap.Duration("duration", time.Since(now)),
			zap.Error(err),
		)
	}()

	var (
		tsFiles []ioutil.TSRangeFile
	)
	if tsFiles, err = ioutil.ListTSRangeFilesInGCDir(
		ctx, c.fs,
	); err != nil || len(tsFiles) == 0 {
		return
	}

	var (
		maxConsumedStart types.TS
		maxConsumedEnd   types.TS
		maxSnapEnd       types.TS
		maxAcctEnd       types.TS
	)

	// Get effective minMerged
	var snapFile, acctFile string
	for _, meta := range tsFiles {
		if meta.IsSnapshotExt() && maxSnapEnd.LT(meta.GetEnd()) {
			maxSnapEnd = *meta.GetEnd()
			snapFile = meta.GetName()
		}
		if meta.IsAcctExt() && maxAcctEnd.LT(meta.GetEnd()) {
			maxAcctEnd = *meta.GetEnd()
			acctFile = meta.GetName()
		}
		c.mutation.metaFiles[meta.GetName()] = meta
	}

	gcFiles := make([]string, 0)
	for _, meta := range tsFiles {
		if meta.IsSnapshotExt() || meta.IsAcctExt() {
			continue
		}
		if maxConsumedStart.IsEmpty() || maxConsumedStart.LT(meta.GetEnd()) {
			maxConsumedStart = *meta.GetStart()
			maxConsumedEnd = *meta.GetEnd()
			gcFiles = append(gcFiles, meta.GetName())
		}
	}

	// In the normal process, gcFiles is empty, and it is impossible to have snapFile and acctFile,
	// but when upgrading from 1.2 to 1.3, there may be such a situation, so you need to replay table info first
	if acctFile != "" {
		if err = c.mutation.snapshotMeta.ReadTableInfo(
			c.ctx,
			ioutil.MakeGCFullName(acctFile),
			c.fs,
		); err != nil {
			return
		}
	}
	if len(gcFiles) == 0 {
		return
	}
	logger := logutil.Info
	for _, name := range gcFiles {
		start := time.Now()
		window := NewGCWindow(c.mp, c.fs)
		if err = window.ReadTable(
			ctx,
			ioutil.MakeGCFullName(name),
			c.fs,
		); err != nil {
			logger = logutil.Error
		}
		logger(
			"GC-REPLAY-READ-TABLE",
			zap.String("name", name),
			zap.Duration("duration", time.Since(start)),
			zap.Error(err),
		)
		if err != nil {
			return
		}
		c.mutAddScannedLocked(window)
	}
	if snapFile != "" {
		if err = c.mutation.snapshotMeta.ReadMeta(
			ctx,
			ioutil.MakeGCFullName(snapFile),
			c.fs,
		); err != nil {
			return
		}
	}
	ckp := checkpoint.NewCheckpointEntry(
		c.sid, maxConsumedStart, maxConsumedEnd, checkpoint.ET_Incremental,
	)
	c.updateScanWaterMark(ckp)
	compacted := c.checkpointCli.GetCompacted()
	if compacted != nil {
		start := time.Now()
		end := compacted.GetEnd()
		c.updateCheckpointGCWaterMark(&end)

		var ckpReader *logtail.CKPReader
		if ckpReader, err = c.getCkpReader(ctx, compacted); err != nil {
			logutil.Error(
				"GC-REPLAY-COLLECT-ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.Error(err),
				zap.String("checkpoint", compacted.String()),
			)
			return
		}
		var snapshots *logtail.SnapshotInfo
		var pitrs *logtail.PitrInfo
		pitrs, err = c.GetPITRsLocked(ctx)
		if err != nil {
			logutil.Error(
				"GC-REPLAY-GET-PITRS_ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.Duration("duration", time.Since(start)),
				zap.String("checkpoint", compacted.String()),
				zap.Error(err),
			)
			return
		}
		// Get backup protection TS from snapshot (taken at GC start)
		var extraTS types.TS
		protectedTS, isActive := c.getBackupProtectionSnapshot()
		if isActive {
			extraTS = protectedTS
		}

		snapshots, err = c.mutation.snapshotMeta.GetSnapshot(c.ctx, c.sid, c.fs, c.mp, extraTS)
		if err != nil {
			logutil.Error(
				"GC-REPLAY-GET-SNAPSHOT_ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.Duration("duration", time.Since(start)),
				zap.String("checkpoint", compacted.String()),
				zap.Error(err),
			)
			return
		}
		_, sarg, _ := fault.TriggerFault("replay error UT")
		if sarg != "" {
			err = moerr.NewInternalErrorNoCtxf("GC-REPLAY-GET-CHECKPOINT-DATA-ERROR %s", sarg)
			return err
		}
		var ckpBatch *batch.Batch
		if ckpBatch, err = ckpReader.GetCheckpointData(ctx); err != nil {
			logutil.Error(
				"GC-REPLAY-GET-CHECKPOINT-DATA_ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.Duration("duration", time.Since(start)),
				zap.String("checkpoint", compacted.String()),
				zap.Error(err),
			)
			return
		}
		defer ckpBatch.Clean(common.CheckpointAllocator)
		logtail.FillUsageBatOfCompacted(
			ctx,
			c.checkpointCli.GetCatalog().GetUsageMemo().(*logtail.TNUsageMemo),
			ckpBatch,
			c.mutation.snapshotMeta,
			snapshots,
			pitrs,
			0)
		logutil.Info(
			"GC-REPLAY-COLLECT-SNAPSHOT-SIZE",
			zap.String("task", c.TaskNameLocked()),
			zap.Duration("duration", time.Since(start)),
			zap.String("checkpoint", compacted.String()),
			zap.Int("count", ckpBatch.RowCount()),
		)
	}
	return
}

func (c *checkpointCleaner) GetCheckpointMetaFiles() map[string]struct{} {
	return c.checkpointCli.GetCheckpointMetaFiles()
}

func (c *checkpointCleaner) updateScanWaterMark(e *checkpoint.CheckpointEntry) {
	c.watermarks.scanWaterMark.Store(e)
}

func (c *checkpointCleaner) updateGCWaterMark(e *checkpoint.CheckpointEntry) {
	c.watermarks.gcWaterMark.Store(e)
}

func (c *checkpointCleaner) updateCheckpointGCWaterMark(ts *types.TS) {
	c.watermarks.checkpointGCWaterMark.Store(ts)
}

func (c *checkpointCleaner) mutAddScannedLocked(window *GCWindow) {
	if c.mutation.scanned == nil {
		c.mutation.scanned = window
	} else {
		c.mutation.scanned.Merge(window)
		window.Close()
	}
}

func (c *checkpointCleaner) GetScanWaterMark() *checkpoint.CheckpointEntry {
	return c.watermarks.scanWaterMark.Load()
}

func (c *checkpointCleaner) GetMinMerged() *checkpoint.CheckpointEntry {
	return c.GetScanWaterMark()
}

func (c *checkpointCleaner) GetGCWaterMark() *checkpoint.CheckpointEntry {
	return c.watermarks.gcWaterMark.Load()
}

func (c *checkpointCleaner) GetCheckpointGCWaterMark() *types.TS {
	return c.watermarks.checkpointGCWaterMark.Load()
}

func (c *checkpointCleaner) GetScannedWindow() *GCWindow {
	c.mutation.Lock()
	defer c.mutation.Unlock()
	return c.mutation.scanned
}

func (c *checkpointCleaner) GetScannedWindowLocked() *GCWindow {
	return c.mutation.scanned
}

func (c *checkpointCleaner) CloneMetaFilesLocked() map[string]ioutil.TSRangeFile {
	metaFiles := make(map[string]ioutil.TSRangeFile, len(c.mutation.metaFiles))
	for k, v := range c.mutation.metaFiles {
		metaFiles[k] = v
	}
	return metaFiles
}

func (c *checkpointCleaner) deleteStaleSnapshotFilesLocked() error {
	var (
		maxSnapEnd  types.TS
		maxSnapFile string

		maxAcctEnd  types.TS
		maxAcctFile string

		err error
	)

	metaFiles := c.CloneMetaFilesLocked()

	prevNum := len(metaFiles)

	doDeleteFileFn := func(
		thisFile string, thisTS *types.TS,
		maxFile string, maxTS *types.TS,
	) (
		newMaxFile string,
		newMaxTS types.TS,
		err error,
	) {
		if maxFile == "" {
			newMaxFile = thisFile
			newMaxTS = *thisTS
			logutil.Info(
				"GC-TRACE-DELETE-SNAPSHOT-FILE",
				zap.String("task", c.TaskNameLocked()),
				zap.String("max-file", newMaxFile),
				zap.String("max-ts", newMaxTS.ToString()),
			)
			return
		}
		if maxTS.LT(thisTS) {
			newMaxFile = thisFile
			newMaxTS = *thisTS
			if err = c.fs.Delete(c.ctx, ioutil.MakeGCFullName(maxFile)); err != nil {
				logutil.Error(
					"GC-DELETE-SNAPSHOT-FILE-ERROR",
					zap.String("task", c.TaskNameLocked()),
					zap.String("file", maxFile),
					zap.Error(err),
					zap.String("new-max-file", newMaxFile),
					zap.String("new-max-ts", newMaxTS.ToString()),
				)
				return
			}
			logutil.Info(
				"GC-TRACE-DELETE-SNAPSHOT-FILE",
				zap.String("task", c.TaskNameLocked()),
				zap.String("max-file", newMaxFile),
				zap.String("max-ts", newMaxTS.ToString()),
			)
			// TODO: seem to be a bug
			delete(metaFiles, maxFile)
			return
		}

		// thisTS <= maxTS: this file is expired and should be deleted
		if err = c.fs.Delete(c.ctx, ioutil.MakeGCFullName(thisFile)); err != nil {
			logutil.Error(
				"GC-DELETE-SNAPSHOT-FILE-ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.String("file", ioutil.MakeGCFullName(thisFile)),
				zap.Error(err),
				zap.String("max-file", maxFile),
				zap.String("max-ts", maxTS.ToString()),
			)
		}
		logutil.Info(
			"GC-TRACE-DELETE-SNAPSHOT-FILE",
			zap.String("task", c.TaskNameLocked()),
			zap.String("max-file", thisFile),
			zap.String("max-ts", thisTS.ToString()),
		)
		delete(metaFiles, thisFile)

		return
	}

	for _, metaFile := range metaFiles {
		switch metaFile.GetExt() {
		case ioutil.SnapshotExt:
			if maxSnapFile, maxSnapEnd, err = doDeleteFileFn(
				metaFile.GetName(), metaFile.GetEnd(), maxSnapFile, &maxSnapEnd,
			); err != nil {
				return err
			}
		case ioutil.AcctExt:
			if maxAcctFile, maxAcctEnd, err = doDeleteFileFn(
				metaFile.GetName(), metaFile.GetEnd(), maxAcctFile, &maxAcctEnd,
			); err != nil {
				return err
			}
		}
	}
	if len(metaFiles) != prevNum {
		var w bytes.Buffer
		for _, v := range metaFiles {
			w.WriteString(fmt.Sprintf("%s,", v.AsString("GCFile")))
		}
		logutil.Info(
			"GC-TRACE-DELETE-SNAPSHOT-FILES",
			zap.String("task", c.TaskNameLocked()),
			zap.Int("left-len", len(metaFiles)),
			zap.String("left-files", w.String()),
		)
	}

	return c.mutSetNewMetaFilesLocked(metaFiles)
}

// when call this function: at least one incremental checkpoint has been scanned
// it uses `c.mutation.metaFiles` and `c.mutation.scanned` as the input
// it updates `c.mutation.metaFiles` after the deletion
// it deletes the stale checkpoint meta files
// Example:
// Before:
// `c.mutation.metaFiles`: [t100_t200_xxx.ckp, t200_t300_xxx.ckp, t300_t400_xxx.ckp]
// `c.mutation.scanned`: [t300, t400]
// After:
// `c.mutation.metaFiles`: [t300_t400_xxx.ckp]
// `c.mutation.scanned`: [t300, t400]
// `t100_t200_xxx.ckp`, `t200_t300_xxx.ckp` are hard deleted
func (c *checkpointCleaner) deleteStaleCKPMetaFileLocked() (err error) {
	// TODO: add log
	window := c.GetScannedWindowLocked()
	metaFiles := c.CloneMetaFilesLocked()
	filesToDelete := make([]string, 0)
	for _, metaFile := range metaFiles {
		if !metaFile.IsCKPFile() ||
			(metaFile.RangeEqual(&window.tsRange.start, &window.tsRange.end)) {
			logutil.Info(
				"GC-TRACE-DELETE-CKP-FILE-SKIP",
				zap.String("task", c.TaskNameLocked()),
				zap.String("skip-file", metaFile.GetName()),
			)
			continue
		}
		gcWindow := NewGCWindow(c.mp, c.fs)
		defer gcWindow.Close()
		if err = gcWindow.ReadTable(
			c.ctx,
			metaFile.GetGCFullName(),
			c.fs,
		); err != nil {
			logutil.Error(
				"GC-WINDOW-READ-ERROR",
				zap.Error(err),
				zap.String("file", metaFile.GetGCFullName()),
				zap.String("task", c.TaskNameLocked()),
			)
			return
		}
		for _, file := range gcWindow.files {
			filesToDelete = append(filesToDelete, file.ObjectName().String())
		}
		filesToDelete = append(filesToDelete, metaFile.GetGCFullName())
		delete(metaFiles, metaFile.GetName())
	}

	// TODO: if file is not found, it should be ignored
	if err = c.fs.Delete(c.ctx, filesToDelete...); err != nil {
		logutil.Error(
			"GC-DELETE-CKP-FILES-ERROR",
			zap.Error(err),
			zap.Strings("files", filesToDelete),
			zap.String("task", c.TaskNameLocked()),
		)
		return
	}
	return c.mutSetNewMetaFilesLocked(metaFiles)
}

// getEntriesToMerge returns the checkpoint entries to merge
func (c *checkpointCleaner) getEntriesToMerge(lowWaterMark *types.TS) (
	entries []*checkpoint.CheckpointEntry,
) {
	gcWaterMark := c.GetCheckpointGCWaterMark()
	start := types.TS{}
	if gcWaterMark != nil {
		start = *gcWaterMark
	}
	if !lowWaterMark.GE(&start) {
		logutil.Warn("GC-PANIC-MERGE-CKP",
			zap.String("task", c.TaskNameLocked()),
			zap.String("start", start.ToString()),
			zap.String("end", lowWaterMark.ToString()),
		)
		return
	}
	compacted := c.checkpointCli.GetCompacted()
	ickps := c.checkpointCli.ICKPRange(&start, lowWaterMark, c.config.maxMergeCheckpointCount)
	if compacted != nil && len(ickps) > 0 {
		entries = make([]*checkpoint.CheckpointEntry, 0, 1+len(ickps))
		entries = append(entries, compacted)
		entries = append(entries, ickps...)
	} else {
		entries = ickps
	}

	logutil.Info(
		"GC-TRACE-GET-META-FILES-TO-MERGE",
		zap.String("task", c.TaskNameLocked()),
		zap.String("start", start.ToString()),
		zap.String("end", lowWaterMark.ToString()),
		zap.Int("checkpoint-len", len(entries)),
	)
	return
}

// filterCheckpoints filters the checkpoints with the endTS less than the highWater
func (c *checkpointCleaner) filterCheckpoints(
	highWater *types.TS,
	checkpoints []*checkpoint.CheckpointEntry,
) ([]*checkpoint.CheckpointEntry, error) {
	if len(checkpoints) == 0 {
		return nil, nil
	}
	var i int
	for i = len(checkpoints) - 1; i >= 0; i-- {
		endTS := checkpoints[i].GetEnd()
		if endTS.LE(highWater) {
			logutil.Info(
				"GC-Filter-CKP",
				zap.String("end-ts", endTS.ToString()),
				zap.String("high-water", highWater.ToString()),
			)
			break
		}
	}
	return checkpoints[:i+1], nil
}

func (c *checkpointCleaner) mergeCheckpointFilesLocked(
	ctx context.Context,
	checkpointLowWaterMark *types.TS,
	memoryBuffer *containers.OneSchemaBatchBuffer,
	snapshots *logtail.SnapshotInfo,
	pitrs *logtail.PitrInfo,
	gcFileCount int,
) (err error) {
	// checkpointLowWaterMark is empty only in the following cases:
	// 1. no incremental and no gloabl checkpoint
	// 2. one incremental checkpoint with empty start
	// 3. no incremental checkpoint and one or more global checkpoints
	// if there are global checkpoints and incremental checkpoints, the low water mark is:
	// min(min(startTS of all incremental checkpoints),min(endTS of global checkpoints))
	if checkpointLowWaterMark.IsEmpty() {
		return
	}

	now := time.Now()

	var (
		deleteFiles       []string
		tmpNewFiles       []string
		newCkp            *checkpoint.CheckpointEntry
		ckpMaxEnd         types.TS
		toMergeCheckpoint []*checkpoint.CheckpointEntry
		extraErrMsg       string
		newCkpData        *batch.Batch
	)

	defer func() {
		if newCkpData != nil {
			newCkpData.Clean(common.CheckpointAllocator)
		}
		logger := logutil.Info
		if err != nil {
			logger = logutil.Error
		}
		logger(
			"GC-TRACE-MERGE-CHECKPOINT-FILES",
			zap.String("task", c.TaskNameLocked()),
			zap.Duration("duration", time.Since(now)),
			zap.Error(err),
			zap.String("extra-err-msg", extraErrMsg),
			zap.Strings("delete-files", deleteFiles),
			zap.String("new-checkpoint", newCkp.String()),
			zap.String("checkpoint-max-end", ckpMaxEnd.ToString()),
			zap.Int("checkpoint-to-merge-len", len(toMergeCheckpoint)),
		)
	}()

	if toMergeCheckpoint = c.getEntriesToMerge(checkpointLowWaterMark); len(toMergeCheckpoint) == 0 {
		return
	}

	ckpMaxEnd = toMergeCheckpoint[len(toMergeCheckpoint)-1].GetEnd()
	if ckpMaxEnd.GT(checkpointLowWaterMark) {
		logutil.Warn("GC-PANIC-MERGE-FILES",
			zap.String("task", c.TaskNameLocked()),
			zap.String("ckpMaxEnd", ckpMaxEnd.ToString()),
			zap.String("checkpointLowWaterMark", checkpointLowWaterMark.ToString()),
		)
		return
	}

	if toMergeCheckpoint, err = c.filterCheckpoints(
		checkpointLowWaterMark,
		toMergeCheckpoint,
	); err != nil {
		return
	}

	if len(toMergeCheckpoint) == 0 {
		return
	}

	// get the scanned window, it should not be nil
	window := c.GetScannedWindowLocked()
	if ckpMaxEnd.GT(&window.tsRange.end) {
		logutil.Warn("GC-PANIC-MERGE-FILES",
			zap.String("ckpMaxEnd", ckpMaxEnd.ToString()),
			zap.String("window-end", window.tsRange.end.ToString()),
		)
		return
	}

	sourcer := window.MakeFilesReader(ctx, c.fs)
	bf, err := BuildBloomfilter(
		ctx,
		Default_Coarse_EstimateRows,
		Default_Coarse_Probility,
		0,
		sourcer.Read,
		memoryBuffer,
		c.mp,
	)

	if deleteFiles, tmpNewFiles, newCkp, newCkpData, err = MergeCheckpoint(
		ctx,
		c.TaskNameLocked(),
		c.sid,
		toMergeCheckpoint,
		bf,
		&ckpMaxEnd,
		c.checkpointCli,
		c.mp,
		c.fs,
	); err != nil {
		extraErrMsg = "MergeCheckpoint failed"
		return err
	}
	logtail.FillUsageBatOfCompacted(
		ctx,
		c.checkpointCli.GetCatalog().GetUsageMemo().(*logtail.TNUsageMemo),
		newCkpData,
		c.mutation.snapshotMeta,
		snapshots,
		pitrs,
		gcFileCount)
	if newCkp == nil {
		logutil.Warn("GC-PANIC-NEW-CHECKPOINT-EMPTY",
			zap.String("task", c.TaskNameLocked()))
		return
	}
	newFiles := tmpNewFiles
	for _, stats := range c.GetScannedWindowLocked().files {
		newFiles = append(newFiles, stats.ObjectName().String())
	}

	if err = c.appendFilesToWAL(newFiles...); err != nil {
		logutil.Error(
			"GC-TRACE-MERGE-CHECKPOINT-FILES",
			zap.String("task", c.TaskNameLocked()),
			zap.Int("file len", len(newFiles)),
			zap.Error(err),
		)
		return
	}

	// SJW TODO: need to handle not updated scenario
	c.checkpointCli.UpdateCompacted(newCkp)

	// update checkpoint gc water mark
	newWaterMark := newCkp.GetEnd()
	c.updateCheckpointGCWaterMark(&newWaterMark)

	gCkps := c.checkpointCli.GetAllGlobalCheckpoints()
	for _, ckp := range gCkps {
		end := ckp.GetEnd()
		logutil.Info(
			"GC-TRACE-GLOBAL-CHECKPOINT-FILE",
			zap.String("task", c.TaskNameLocked()),
			zap.String("gckp", ckp.String()),
		)
		if end.LT(&newWaterMark) {
			nameMeta := ioutil.EncodeCKPMetadataFullName(
				ckp.GetStart(), ckp.GetEnd(),
			)
			logutil.Info(
				"GC-TRACE-DELETE-GLOBAL-CHECKPOINT-FILE",
				zap.String("task", c.TaskNameLocked()),
				zap.String("gckp", nameMeta),
			)

			deleteFiles = append(deleteFiles, nameMeta)
			// fix gckp files leak
			var files []string
			files, err = getCheckpointLocation(ctx, ckp, c.fs)
			if err != nil {
				extraErrMsg = "getCheckpointLocation failed"
				return err
			}
			if len(files) > 0 {
				deleteFiles = append(deleteFiles, files...)
			}

		}
	}
	if c.GCCheckpointEnabled() {
		if err = c.fs.Delete(ctx, deleteFiles...); err != nil {
			extraErrMsg = "DelFiles failed"
			return err
		}
	}

	for _, deleteFile := range deleteFiles {
		_, decodedFile := ioutil.TryDecodeTSRangeFile(deleteFile)
		if decodedFile.IsMetadataFile() || decodedFile.IsCompactExt() {
			logutil.Info(
				"GC-TRACE-DELETE-CHECKPOINT-FILE",
				zap.String("task", c.TaskNameLocked()),
				zap.String("delete file", deleteFile),
			)
			c.checkpointCli.RemoveCheckpointMetaFile(decodedFile.GetName())
		}
	}
	return nil
}

func (c *checkpointCleaner) getCkpReader(
	ctx context.Context,
	ckp *checkpoint.CheckpointEntry,
) (reader *logtail.CKPReader, err error) {
	return logtail.GetCheckpointReader(
		ctx, c.sid, c.fs, ckp.GetLocation(), ckp.GetVersion(),
	)
}

func (c *checkpointCleaner) GetPITRs() (*logtail.PitrInfo, error) {
	c.mutation.Lock()
	defer c.mutation.Unlock()
	ts := time.Now()
	return c.mutation.snapshotMeta.GetPITR(c.ctx, c.sid, ts, c.fs, c.mp)
}

func (c *checkpointCleaner) GetPITRsLocked(ctx context.Context) (*logtail.PitrInfo, error) {
	ts := time.Now()
	return c.mutation.snapshotMeta.GetPITR(ctx, c.sid, ts, c.fs, c.mp)
}

// (no incremental checkpoint scan)
// `tryGCLocked` will update
// `mutation.scanned` and `mutation.metaFiles` and `mutation.snapshotMeta`
// it will update the GC watermark and the checkpoint GC watermark
// `mutation.scanned`: it will be GC'ed against the max global checkpoint.
func (c *checkpointCleaner) tryGCLocked(
	ctx context.Context,
	memoryBuffer *containers.OneSchemaBatchBuffer,
) (err error) {
	// 1. Quick check if GC is needed
	// 1.1. If there is no global checkpoint, no need to do GC
	var maxGlobalCKP *checkpoint.CheckpointEntry
	if maxGlobalCKP = c.checkpointCli.MaxGlobalCheckpoint(); maxGlobalCKP == nil {
		return
	}
	// 1.2. If there is no incremental checkpoint scanned, no need to do GC.
	//      because GC is based on the scanned result.
	var scannedWindow *GCWindow
	if scannedWindow = c.GetScannedWindowLocked(); scannedWindow == nil {
		return
	}

	// gcWaterMark is not nil, which means some global checkpoint has been GC'ed
	// if the GC'ed global checkpoint is greater than or equal to the max global checkpoint,
	// it means no need to do GC again
	gcWaterMark := c.GetGCWaterMark()
	if gcWaterMark != nil {
		gcWaterMarkTS := gcWaterMark.GetEnd()
		maxGlobalCKPTS := maxGlobalCKP.GetEnd()
		if gcWaterMarkTS.GE(&maxGlobalCKPTS) {
			return
		}
	}

	if err = c.tryGCAgainstGCKPLocked(
		ctx, maxGlobalCKP, memoryBuffer,
	); err != nil {
		logutil.Error(
			"GC-TRY-GC-AGAINST-GCKP-ERROR",
			zap.Error(err),
			zap.String("checkpoint", maxGlobalCKP.String()),
			zap.String("task", c.TaskNameLocked()),
		)
		return
	}

	if err = c.deleteStaleCKPMetaFileLocked(); err != nil {
		logutil.Error(
			"GC-TRY-DELETE-STALE-CKP-META-FILE-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
			zap.String("checkpoint", maxGlobalCKP.String()),
		)
	}

	if err = c.deleteStaleSnapshotFilesLocked(); err != nil {
		logutil.Error(
			"GC-TRY-DELETE-STALE-SNAPSHOT-FILES-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
			zap.String("checkpoint", maxGlobalCKP.String()),
		)
	}
	return
}

// when calling this function:
// at least one incremental checkpoint has been scanned
// the GC'ed water mark less than the global checkpoint
// `gckp` is the global checkpoint that needs to be GC'ed against
// `memoryBuffer` is the buffer used to read the data of the GC window
func (c *checkpointCleaner) tryGCAgainstGCKPLocked(
	ctx context.Context,
	gckp *checkpoint.CheckpointEntry,
	memoryBuffer *containers.OneSchemaBatchBuffer,
) (err error) {
	now := time.Now()
	var snapshots *logtail.SnapshotInfo
	var extraErrMsg string
	defer func() {
		logutil.Info(
			"GC-TRACE-TRY-GC-AGAINST-GCKP",
			zap.String("task", c.TaskNameLocked()),
			zap.Duration("duration", time.Since(now)),
			zap.String("checkpoint", gckp.String()),
			zap.Error(err),
			zap.String("extra-err-msg", extraErrMsg),
		)
	}()
	pitrs, err := c.GetPITRsLocked(ctx)
	if err != nil {
		extraErrMsg = "GetPITRs failed"
		return
	}

	// Use snapshot taken at GC start to ensure consistency
	var extraTS types.TS
	protectedTS, isActive := c.getBackupProtectionSnapshot()
	if isActive {
		extraTS = protectedTS
	}

	snapshots, err = c.mutation.snapshotMeta.GetSnapshot(ctx, c.sid, c.fs, c.mp, extraTS)
	if err != nil {
		extraErrMsg = "GetSnapshot failed"
		return
	}
	filesToGC, err := c.doGCAgainstGlobalCheckpointLocked(
		ctx, gckp, snapshots, pitrs, memoryBuffer,
	)
	if err != nil {
		extraErrMsg = "doGCAgainstGlobalCheckpointLocked failed"
		return
	}
	// Delete files after doGCAgainstGlobalCheckpointLocked
	// TODO:Requires Physical Removal Policy
	// Note: Data files are GC'ed normally even when backup protection is active.
	// Only checkpoint metadata merge/delete is skipped (handled in mergeCheckpointFilesLocked).
	if err = c.deleter.DeleteMany(
		ctx,
		c.TaskNameLocked(),
		filesToGC,
	); err != nil {
		extraErrMsg = fmt.Sprintf("ExecDelete %v failed", filesToGC)
		return
	}
	if c.GetGCWaterMark() == nil {
		return nil
	}
	waterMark := c.GetGCWaterMark().GetEnd()
	mergeMark := types.TS{}
	if c.GetCheckpointGCWaterMark() != nil {
		mergeMark = *c.GetCheckpointGCWaterMark()
	}
	if !waterMark.GT(&mergeMark) {
		return nil
	}
	scanMark := c.GetScanWaterMark().GetEnd()
	if scanMark.IsEmpty() {
		logutil.Warn("GC-PANIC-SCANMARK-EMPTY",
			zap.String("task", c.TaskNameLocked()),
			zap.String("mergeMark", mergeMark.ToString()))
		return nil
	}
	if waterMark.GT(&scanMark) {
		waterMark = scanMark
	}
	err = c.mergeCheckpointFilesLocked(
		ctx, &waterMark, memoryBuffer, snapshots, pitrs, len(filesToGC),
	)
	if err != nil {
		extraErrMsg = fmt.Sprintf("mergeCheckpointFilesLocked %v failed", waterMark.ToString())
	}
	return
}

// at least one incremental checkpoint has been scanned
// and the GC'ed water mark less than the global checkpoint
func (c *checkpointCleaner) doGCAgainstGlobalCheckpointLocked(
	ctx context.Context,
	gckp *checkpoint.CheckpointEntry,
	snapshots *logtail.SnapshotInfo,
	pitrs *logtail.PitrInfo,
	memoryBuffer *containers.OneSchemaBatchBuffer,
) ([]string, error) {
	now := time.Now()

	var (
		filesToGC                   []string
		metafile                    string
		err                         error
		softDuration, mergeDuration time.Duration
		extraErrMsg                 string
		iscp                        map[uint64]types.TS
	)

	defer func() {
		logutil.Info(
			"GC-TRACE-DO-GC-AGAINST-GCKP",
			zap.String("task", c.TaskNameLocked()),
			zap.Duration("duration", time.Since(now)),
			zap.Duration("soft-gc", softDuration),
			zap.Duration("merge-table", mergeDuration),
			zap.Error(err),
			zap.String("metafile", metafile),
			zap.String("extra-err-msg", extraErrMsg),
		)
	}()

	// do GC against the global checkpoint
	// the result is the files that need to be deleted
	// it will update the file list in the oneWindow
	// Before:
	// [t100, t400] [f1, f2, f3, f4, f5, f6, f7, f8, f9]
	// After:
	// [t100, t400] [f10, f11]
	// Also, it will update the GC metadata
	scannedWindow := c.GetScannedWindowLocked()
	iscp, err = c.ISCPTables()
	if err != nil {
		return nil, err
	}
	if filesToGC, metafile, err = scannedWindow.ExecuteGlobalCheckpointBasedGC(
		ctx,
		gckp,
		snapshots,
		pitrs,
		c.mutation.snapshotMeta,
		iscp,
		c.checkpointCli,
		memoryBuffer,
		c.config.canGCCacheSize,
		c.config.estimateRows,
		c.config.probility,
		c.mp,
		c.fs,
	); err != nil {
		extraErrMsg = fmt.Sprintf("ExecuteGlobalCheckpointBasedGC %v failed", gckp.String())
		return nil, err
	}

	if err = c.appendFilesToWAL(
		ioutil.MakeFullName(scannedWindow.dir, metafile),
	); err != nil {
		logutil.Error(
			"GC-TRACE-APPEND-FILES-TO-WAL-FAILED",
			zap.String("task", c.TaskNameLocked()),
			zap.String("metafile", metafile),
			zap.Error(err))
		return nil, err
	}
	c.mutAddMetaFileLocked(metafile, ioutil.NewTSRangeFile(
		metafile,
		ioutil.CheckpointExt,
		scannedWindow.tsRange.start,
		scannedWindow.tsRange.end,
	))
	softDuration = time.Since(now)

	// update gc watermark and refresh snapshot meta with the latest gc result
	// gcWaterMark will be updated to the end of the global checkpoint after each GC
	// Before:
	// gcWaterMark: GCKP[t100, t200)
	// After:
	// gcWaterMark: GCKP[t200, t400)
	now = time.Now()
	// TODO:
	c.updateGCWaterMark(gckp)
	c.mutation.snapshotMeta.MergeTableInfo(snapshots, pitrs)
	mergeDuration = time.Since(now)
	return filesToGC, nil
}

func (c *checkpointCleaner) scanCheckpointsAsDebugWindow(
	ckps []*checkpoint.CheckpointEntry,
	buffer *containers.OneSchemaBatchBuffer,
) (window *GCWindow, err error) {
	window = NewGCWindow(c.mp, c.fs, WithWindowDir("debug/"))
	if _, err = window.ScanCheckpoints(
		c.ctx, ckps, c.getCkpReader, nil, nil, buffer,
	); err != nil {
		window.Close()
		window = nil
	}
	return
}

func (c *checkpointCleaner) Verify(ctx context.Context) string {
	c.StartMutationTask("gc-verify")
	defer c.StopMutationTask()
	checker := &gcChecker{
		cleaner: c,
	}
	return checker.Verify(ctx, c.mp)
}

func (c *checkpointCleaner) DoCheck(ctx context.Context) error {
	c.StartMutationTask("gc-do-check")
	defer c.StopMutationTask()

	debugCandidates := c.checkpointCli.GetAllIncrementalCheckpoints()
	cpt := c.checkpointCli.GetCompacted()
	gckps := c.checkpointCli.GetAllGlobalCheckpoints()
	// no scan watermark, GC has not yet run
	var scanWaterMark *checkpoint.CheckpointEntry
	if scanWaterMark = c.GetScanWaterMark(); scanWaterMark == nil {
		return moerr.NewInternalErrorNoCtx("GC has not yet run")
	}

	gCkp := c.GetGCWaterMark()
	testutils.WaitExpect(10000, func() bool {
		gCkp = c.GetGCWaterMark()
		return gCkp != nil
	})
	if gCkp == nil {
		gCkp = c.checkpointCli.MaxGlobalCheckpoint()
		if gCkp == nil {
			return nil
		}
		logutil.Warnf("MaxCompared is nil, use maxGlobalCkp %v", gCkp.String())
	}

	for i, checkpoint := range debugCandidates {
		maxEnd := scanWaterMark.GetEnd()
		ckpEnd := checkpoint.GetEnd()
		if ckpEnd.Equal(&maxEnd) {
			debugCandidates = debugCandidates[:i+1]
			break
		}
	}

	start1 := debugCandidates[len(debugCandidates)-1].GetEnd()
	start2 := scanWaterMark.GetEnd()
	if !start1.Equal(&start2) {
		err := moerr.NewInternalErrorNoCtx("TS Compare not equal")
		logutil.Error(
			"GC-COMPRE-ERROR",
			zap.String("task", c.TaskNameLocked()),
			zap.Error(err),
			zap.String("start1", start1.ToString()),
			zap.String("start2", start2.ToString()),
		)
		return err
	}
	buffer := MakeGCWindowBuffer(16 * mpool.MB)
	defer buffer.Close(c.mp)

	debugWindow, err := c.scanCheckpointsAsDebugWindow(
		debugCandidates, buffer,
	)
	if err != nil {
		logutil.Error(
			"GC-SCAN-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
			zap.String("checkpoint", debugCandidates[0].String()),
		)
		// TODO
		return err
	}
	var snapshots *logtail.SnapshotInfo
	snapshots, err = c.GetSnapshotsLocked()
	if err != nil {
		logutil.Error(
			"GC-GET-SNAPSHOTS-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
		)
		return err
	}
	var pitr *logtail.PitrInfo
	pitr, err = c.GetPITRsLocked(c.ctx)
	if err != nil {
		logutil.Error(
			"GC-GET-PITR-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
		)
		return err
	}

	mergeWindow := c.GetScannedWindowLocked().Clone()
	defer mergeWindow.Close()
	logutil.Info(
		"GC-TRACE-MERGE-WINDOW",
		zap.String("task", c.TaskNameLocked()),
		zap.Int("files-count", len(mergeWindow.files)),
	)
	iscp, err := c.ISCPTables()
	if err != nil {
		return err
	}
	if _, _, err = mergeWindow.ExecuteGlobalCheckpointBasedGC(
		c.ctx,
		gCkp,
		snapshots,
		pitr,
		c.mutation.snapshotMeta,
		iscp,
		c.checkpointCli,
		buffer,
		c.config.canGCCacheSize,
		c.config.estimateRows,
		c.config.probility,
		c.mp,
		c.fs,
	); err != nil {
		logutil.Error(
			"GC-EXECUTE-GC-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
		)
		return err
	}

	//logutil.Infof("debug table is %d, stats is %v", len(debugWindow.files.stats), debugWindow.files.stats[0].ObjectName().String())
	if _, _, err = debugWindow.ExecuteGlobalCheckpointBasedGC(
		c.ctx,
		gCkp,
		snapshots,
		pitr,
		c.mutation.snapshotMeta,
		iscp,
		c.checkpointCli,
		buffer,
		c.config.canGCCacheSize,
		c.config.estimateRows,
		c.config.probility,
		c.mp,
		c.fs,
	); err != nil {
		logutil.Error(
			"GC-EXECUTE-GC-ERROR",
			zap.Error(err),
			zap.String("task", c.TaskNameLocked()),
		)
		return err
	}

	//logutil.Infof("debug table2 is %d, stats is %v", len(debugWindow.files.stats), debugWindow.files.stats[0].ObjectName().String())
	objects1, objects2, equal := mergeWindow.Compare(debugWindow, buffer)
	if !equal {
		logutil.Error(
			"GC-WINDOW-COMPARE-ERROR",
			zap.String("task", c.TaskNameLocked()),
			zap.String("merge-window", mergeWindow.String(objects1)),
			zap.String("debug-window", debugWindow.String(objects2)),
		)
		return moerr.NewInternalErrorNoCtx("Compare is failed")
	} else {
		logutil.Info(
			"GC-WINDOW-COMPARE-SUCCESS",
			zap.String("task", c.TaskNameLocked()),
			zap.String("merge-window", mergeWindow.String(objects1)),
			zap.String("debug-window", debugWindow.String(objects2)),
		)
	}

	if cpt == nil {
		return nil
	}
	cptEnd := cpt.GetEnd()
	debugEnd := debugCandidates[0].GetEnd()
	if debugEnd.GT(&cptEnd) {
		return nil
	}
	ickpObjects := make(map[string]map[uint64]*ObjectEntry, 0)
	ok := false
	gcWaterMark := cptEnd
	for _, ckp := range gckps {
		end := ckp.GetEnd()
		if end.GE(&gcWaterMark) {
			gcWaterMark = ckp.GetEnd()
		}
	}
	for i, ckp := range debugCandidates {
		end := ckp.GetEnd()
		end = end.Next()
		if end.Equal(&gcWaterMark) {
			debugCandidates = debugCandidates[:i+1]
			ok = true
			break
		}
	}
	if !ok {
		return nil
	}
	for _, ckp := range debugCandidates {
		ckpReader, err := c.getCkpReader(c.ctx, ckp)
		if err != nil {
			return err
		}
		collectObjectsFromCheckpointData(c.ctx, ckpReader, ickpObjects)
	}
	cptCkpObjects := make(map[string]map[uint64]*ObjectEntry, 0)
	ckpReader, err := c.getCkpReader(c.ctx, cpt)
	if err != nil {
		return err
	}
	collectObjectsFromCheckpointData(c.ctx, ckpReader, cptCkpObjects)

	tList, pList := c.mutation.snapshotMeta.AccountToTableSnapshots(snapshots, pitr)
	for name, tables := range ickpObjects {
		for _, entry := range tables {
			if cptCkpObjects[name] != nil {
				continue
			}
			if logtail.ObjectIsSnapshotRefers(
				entry.stats, pList[entry.table], &entry.createTS, &entry.dropTS, tList[entry.table],
			) {
				if entry.dropTS.IsEmpty() || entry.dropTS.LT(&cptEnd) || (pList[entry.table] != nil && entry.dropTS.GT(pList[entry.table])) {
					continue
				}
				logutil.Error(
					"GC-SNAPSHOT-REFERS-ERROR",
					zap.String("task", c.TaskNameLocked()),
					zap.String("name", entry.stats.ObjectName().String()),
					zap.String("pitr", pList[entry.table].ToString()),
					zap.String("create-ts", entry.createTS.ToString()),
					zap.String("drop-ts", entry.dropTS.ToString()),
				)
				return moerr.NewInternalError(c.ctx, "snapshot refers")
			}
		}
	}
	return nil
}

func (c *checkpointCleaner) Process(
	inputCtx context.Context,
	checker func(*checkpoint.CheckpointEntry) bool) (err error) {
	if !c.GCEnabled() {
		return
	}
	now := time.Now()

	// Start mutation task first to acquire the lock
	// This ensures backup protection operations wait for GC to complete
	c.StartMutationTask("gc-process")
	defer c.StopMutationTask()

	// Check backup protection state and create a snapshot at the start of GC
	// This snapshot will be used throughout the GC process to ensure consistency
	c.backupProtection.Lock()
	if c.backupProtection.isActive && time.Since(c.backupProtection.lastUpdateTime) > 20*time.Minute {
		logutil.Warn(
			"GC-Backup-Protection-Expired-Remove",
			zap.Duration("time-since-update", time.Since(c.backupProtection.lastUpdateTime)),
		)
		c.backupProtection.isActive = false
		c.backupProtection.protectedTS = types.TS{}
	}
	// Create snapshot of backup protection state
	c.mutation.backupProtectionSnapshot.protectedTS = c.backupProtection.protectedTS
	c.mutation.backupProtectionSnapshot.isActive = c.backupProtection.isActive
	isBackupActive := c.backupProtection.isActive
	protectedTS := c.backupProtection.protectedTS
	c.backupProtection.Unlock()

	// If backup protection is active, skip all GC operations
	if isBackupActive {
		logutil.Info(
			"GC-Backup-Protection-Skip-All-GC",
			zap.String("task", c.TaskNameLocked()),
			zap.String("protected-ts", protectedTS.ToString()),
		)
		return nil
	}

	startScanWaterMark := c.GetScanWaterMark()
	startGCWaterMark := c.GetGCWaterMark()

	defer func() {
		endScanWaterMark := c.GetScanWaterMark()
		endGCWaterMark := c.GetGCWaterMark()
		logutil.Info(
			"GC-TRACE-PROCESS",
			zap.String("task", c.TaskNameLocked()),
			zap.Duration("duration", time.Since(now)),
			zap.Error(err),
			zap.String("start-scan-watermark", startScanWaterMark.String()),
			zap.String("end-scan-watermark", endScanWaterMark.String()),
			zap.String("start-gc-watermark", startGCWaterMark.String()),
			zap.String("end-gc-watermark", endGCWaterMark.String()),
		)
	}()

	ctx, cancel := context.WithCancelCause(inputCtx)
	defer cancel(nil)
	go func() {
		select {
		case <-ctx.Done():
		case <-c.ctx.Done():
			cancel(context.Cause(c.ctx))
		case <-inputCtx.Done():
			cancel(context.Cause(inputCtx))
		}
	}()

	select {
	case <-ctx.Done():
		return context.Cause(ctx)
	default:
	}

	memoryBuffer := MakeGCWindowBuffer(16 * mpool.MB)
	defer memoryBuffer.Close(c.mp)

	var tryGC bool
	if err, tryGC = c.tryScanLocked(ctx, memoryBuffer, checker); err != nil {
		return
	}
	if !tryGC {
		return
	}
	err = c.tryGCLocked(ctx, memoryBuffer)
	return
}

// tryScanLocked scans the incremental checkpoints and tries to create a new GC window
// it will update `mutation.scanned` and `mutation.metaFiles`
// it will update the scan watermark
// it will save the snapshot meta and table info to the disk
func (c *checkpointCleaner) tryScanLocked(
	ctx context.Context,
	memoryBuffer *containers.OneSchemaBatchBuffer,
	checker func(*checkpoint.CheckpointEntry) bool,
) (err error, tryGC bool) {

	tryGC = true
	// get the max scanned timestamp
	var maxScannedTS types.TS
	if scanWaterMark := c.GetScanWaterMark(); scanWaterMark != nil {
		maxScannedTS = scanWaterMark.GetEnd()
	}
	candidates := make([]*checkpoint.CheckpointEntry, 0)
	if maxScannedTS.IsEmpty() {
		maxGCkp := c.checkpointCli.MaxGlobalCheckpoint()
		minCkp := c.checkpointCli.MinIncrementalCheckpoint()
		var start types.TS
		if minCkp != nil {
			start = minCkp.GetStart()
		}
		if !start.IsEmpty() && maxGCkp != nil {
			maxScannedTS = maxGCkp.GetEnd()
			cpt := c.checkpointCli.GetCompacted()
			if cpt == nil {
				logutil.Info("GC-PANIC-REBUILD-TABLE",
					zap.String("max gCkp", maxGCkp.String()),
					zap.String("start", start.ToString()))
			} else {
				candidates = append(candidates, cpt)
			}
			candidates = append(candidates, maxGCkp)
			gcWaterMark := c.GetGCWaterMark()
			if gcWaterMark != nil {
				logutil.Warn("GC-PANIC-REBUILD-GC-WATER-MARK",
					zap.String("max gCkp", maxGCkp.String()),
					zap.String("gcWaterMark", gcWaterMark.String()))
			}
			c.updateGCWaterMark(maxGCkp)
			tryGC = false
		}
	}

	// get up to maxMergeCheckpointCount incremental checkpoints starting from the max scanned timestamp
	ckps := c.checkpointCli.ICKPSeekLT(maxScannedTS, c.config.maxMergeCheckpointCount)

	// quick return if there is no incremental checkpoint
	if len(ckps) == 0 && len(candidates) == 0 {
		return
	}

	// filter out the incremental checkpoints that do not meet the requirements
	for _, ckp := range ckps {
		if checker != nil {
			if !checker(ckp) {
				continue
			}
		} else {
			if !c.checkExtras(ckp) {
				continue
			}
		}
		candidates = append(candidates, ckp)
	}

	if len(candidates) == 0 {
		return
	}

	var window *GCWindow
	var tmpNewFiles []string
	if window, tmpNewFiles, err = c.scanCheckpointsLocked(
		ctx, candidates, memoryBuffer,
	); err != nil {
		logutil.Error(
			"GC-SCAN-WINDOW-ERROR",
			zap.Error(err),
			zap.String("checkpoint", candidates[0].String()),
		)
		return
	}
	c.mutAddScannedLocked(window)
	c.updateScanWaterMark(candidates[len(candidates)-1])
	files := tmpNewFiles
	for _, stats := range c.GetScannedWindowLocked().files {
		files = append(files, stats.ObjectName().String())
	}
	if err = c.appendFilesToWAL(files...); err != nil {
		logutil.Error(
			"GC-APPEND-SNAPSHOT-TO-WAL-ERROR",
			zap.String("task", c.TaskNameLocked()),
			zap.Error(err),
		)
		return
	}
	return
}

func (c *checkpointCleaner) mutSetNewMetaFilesLocked(
	metaFiles map[string]ioutil.TSRangeFile,
) error {
	c.mutation.metaFiles = metaFiles
	return nil
}

func (c *checkpointCleaner) mutAddMetaFileLocked(
	key string,
	metaFile ioutil.TSRangeFile,
) error {
	c.mutation.metaFiles[key] = metaFile
	return nil
}

func (c *checkpointCleaner) checkExtras(item any) bool {
	// First check backup protection
	if !c.checkBackupProtection(item) {
		return false
	}

	c.checker.RLock()
	defer c.checker.RUnlock()
	for _, checker := range c.checker.extras {
		if !checker(item) {
			return false
		}
	}
	return true
}

// AddChecker add&update a checker to the cleaner，return the number of checkers
// key is the unique identifier of the checker
func (c *checkpointCleaner) AddChecker(checker func(item any) bool, key string) int {
	c.checker.Lock()
	defer c.checker.Unlock()
	c.checker.extras[key] = checker
	return len(c.checker.extras)
}

// RemoveChecker remove a checker from the cleaner，return true if the checker is removed successfully
func (c *checkpointCleaner) RemoveChecker(key string) error {
	c.checker.Lock()
	defer c.checker.Unlock()
	if len(c.checker.extras) == 1 {
		return moerr.NewCantDelGCCheckerNoCtx()
	}
	delete(c.checker.extras, key)
	return nil
}

// SetBackupProtection sets the backup protection timestamp
// protectedTS is the timestamp that should be protected from GC
// This method only acquires the backupProtection lock, not the mutation lock.
// GC consistency is ensured by Process() which creates a snapshot of protection
// state at GC start, so ongoing GC won't be affected by protection changes.
func (c *checkpointCleaner) SetBackupProtection(protectedTS types.TS) {
	c.backupProtection.Lock()
	defer c.backupProtection.Unlock()
	c.backupProtection.protectedTS = protectedTS
	c.backupProtection.lastUpdateTime = time.Now()
	c.backupProtection.isActive = true
	logutil.Info(
		"GC-Backup-Protection-Set",
		zap.String("protected-ts", protectedTS.ToString()),
		zap.Time("last-update-time", c.backupProtection.lastUpdateTime),
	)
}

// UpdateBackupProtection updates the backup protection timestamp
// This method only acquires the backupProtection lock, not the mutation lock.
// GC consistency is ensured by Process() which creates a snapshot of protection
// state at GC start, so ongoing GC won't be affected by protection changes.
func (c *checkpointCleaner) UpdateBackupProtection(protectedTS types.TS) {
	c.backupProtection.Lock()
	defer c.backupProtection.Unlock()
	if !c.backupProtection.isActive {
		logutil.Warn("GC-Backup-Protection-Update-Not-Active")
		return
	}
	c.backupProtection.protectedTS = protectedTS
	c.backupProtection.lastUpdateTime = time.Now()
	logutil.Info(
		"GC-Backup-Protection-Updated",
		zap.String("protected-ts", protectedTS.ToString()),
		zap.Time("last-update-time", c.backupProtection.lastUpdateTime),
	)
}

// RemoveBackupProtection removes the backup protection
func (c *checkpointCleaner) RemoveBackupProtection() {
	c.backupProtection.Lock()
	defer c.backupProtection.Unlock()
	c.backupProtection.isActive = false
	c.backupProtection.protectedTS = types.TS{}
	logutil.Info("GC-Backup-Protection-Removed")
}

// GetBackupProtection returns the backup protection info
func (c *checkpointCleaner) GetBackupProtection() (protectedTS types.TS, lastUpdateTime time.Time, isActive bool) {
	c.backupProtection.RLock()
	defer c.backupProtection.RUnlock()
	return c.backupProtection.protectedTS, c.backupProtection.lastUpdateTime, c.backupProtection.isActive
}

// getBackupProtectionSnapshot returns the backup protection snapshot taken at the start of GC
// This ensures GC consistency: once GC starts, it uses this snapshot and ignores
// any changes to backup protection during GC execution
func (c *checkpointCleaner) getBackupProtectionSnapshot() (protectedTS types.TS, isActive bool) {
	// Use snapshot from mutation (taken at GC start)
	// No lock needed as mutation is already locked during GC execution
	return c.mutation.backupProtectionSnapshot.protectedTS, c.mutation.backupProtectionSnapshot.isActive
}

// checkBackupProtection checks if the checkpoint should be protected from GC
// Returns true if the checkpoint can be GC'ed, false if it should be protected
// This function uses the snapshot taken at GC start to ensure consistency
func (c *checkpointCleaner) checkBackupProtection(item any) bool {
	// Use snapshot from mutation (taken at GC start)
	protectedTS, isActive := c.getBackupProtectionSnapshot()

	// If backup protection is not active, allow GC
	if !isActive {
		return true
	}

	// Check if the item is a checkpoint entry
	ckp, ok := item.(*checkpoint.CheckpointEntry)
	if !ok {
		// For non-checkpoint items, allow GC (metadata files are checked separately in deleteStaleSnapshotFilesLocked)
		return true
	}

	// For checkpoint entries, check if the end timestamp is less than or equal to protected timestamp
	// We protect checkpoints whose end timestamp is <= protected timestamp
	// This means we protect all checkpoints up to and including the backup time point
	endTS := ckp.GetEnd()
	// Empty/invalid timestamps should not be protected (allow GC)
	if endTS.IsEmpty() {
		return true
	}
	if endTS.LE(&protectedTS) {
		logutil.Info(
			"GC-Backup-Protection-Block-Checkpoint",
			zap.String("checkpoint-end-ts", endTS.ToString()),
			zap.String("protected-ts", protectedTS.ToString()),
			zap.String("checkpoint", ckp.String()),
		)
		return false
	}

	return true
}

// appendFilesToWAL append the GC meta files to WAL.
func (c *checkpointCleaner) appendFilesToWAL(files ...string) error {
	if c.logDriver == nil {
		return nil
	}
	entry, err := wal.BuildFilesEntry(files)
	if err != nil {
		return err
	}
	_, err = c.logDriver.AppendEntry(wal.GroupFiles, entry)
	if err != nil {
		return err
	}
	return nil
}

// this function will update:
// `c.mutation.metaFiles`
// `c.mutation.snapshotMeta`
// this function will save the snapshot meta and table info to the disk
func (c *checkpointCleaner) scanCheckpointsLocked(
	ctx context.Context,
	ckps []*checkpoint.CheckpointEntry,
	memoryBuffer *containers.OneSchemaBatchBuffer,
) (gcWindow *GCWindow, newFiles []string, err error) {
	now := time.Now()

	var (
		snapSize, tableSize uint32
	)
	defer func() {
		logutil.Info(
			"GC-TRACE-SCAN",
			zap.String("task", c.TaskNameLocked()),
			zap.Int("checkpoint-count", len(ckps)),
			zap.Duration("duration", time.Since(now)),
			zap.Uint32("snap-meta-size :", snapSize),
			zap.Uint32("table-meta-size :", tableSize),
			zap.String("snapshot-detail", c.mutation.snapshotMeta.String()))
	}()

	var snapshotFile, accountFile ioutil.TSRangeFile
	newFiles = make([]string, 0, 3)
	saveSnapshot := func() (err2 error) {
		select {
		case <-ctx.Done():
			err2 = context.Cause(ctx)
			return
		default:
		}
		name := ioutil.EncodeSnapshotMetadataName(
			ckps[0].GetStart(), ckps[len(ckps)-1].GetEnd(),
		)
		if snapSize, err2 = c.mutation.snapshotMeta.SaveMeta(
			ioutil.MakeGCFullName(name),
			c.fs,
		); err2 != nil {
			logutil.Error(
				"GC-SAVE-SNAPSHOT-META-ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.Error(err2),
			)
			return
		}
		newFiles = append(newFiles, ioutil.MakeGCFullName(name))
		snapshotFile = ioutil.NewTSRangeFile(
			name,
			ioutil.SnapshotExt,
			ckps[0].GetStart(),
			ckps[len(ckps)-1].GetEnd(),
		)
		name = ioutil.EncodeAcctMetadataName(
			ckps[0].GetStart(), ckps[len(ckps)-1].GetEnd(),
		)
		if tableSize, err2 = c.mutation.snapshotMeta.SaveTableInfo(
			ioutil.MakeGCFullName(name),
			c.fs,
		); err2 != nil {
			logutil.Error(
				"GC-SAVE-TABLE-META-ERROR",
				zap.String("task", c.TaskNameLocked()),
				zap.Error(err2),
			)
		}
		newFiles = append(newFiles, ioutil.MakeGCFullName(name))
		accountFile = ioutil.NewTSRangeFile(
			name,
			ioutil.AcctExt,
			ckps[0].GetStart(),
			ckps[len(ckps)-1].GetEnd(),
		)
		return
	}

	gcWindow = NewGCWindow(c.mp, c.fs)
	var gcMetaFile string
	if gcMetaFile, err = gcWindow.ScanCheckpoints(
		ctx,
		ckps,
		c.getCkpReader,
		c.mutUpdateSnapshotMetaLocked,
		saveSnapshot,
		memoryBuffer,
	); err != nil {
		gcWindow.Close()
		gcWindow = nil
		return
	}
	newFiles = append(newFiles, ioutil.MakeFullName(gcWindow.dir, gcMetaFile))
	c.mutAddMetaFileLocked(snapshotFile.GetName(), snapshotFile)
	c.mutAddMetaFileLocked(accountFile.GetName(), accountFile)
	c.mutAddMetaFileLocked(
		gcMetaFile,
		ioutil.NewTSRangeFile(
			gcMetaFile,
			ioutil.CheckpointExt,
			gcWindow.tsRange.start,
			gcWindow.tsRange.end,
		),
	)
	return
}

func (c *checkpointCleaner) mutUpdateSnapshotMetaLocked(
	ckp *checkpoint.CheckpointEntry,
	data *logtail.CKPReader,
) error {
	return c.mutation.snapshotMeta.Update(
		c.ctx,
		c.fs,
		data,
		ckp.GetStart(),
		ckp.GetEnd(),
		c.TaskNameLocked(),
	)
}

func (c *checkpointCleaner) GetSnapshots() (*logtail.SnapshotInfo, error) {
	c.mutation.Lock()
	defer c.mutation.Unlock()

	// Use snapshot if GC is running, otherwise use current protection state
	var extraTS types.TS
	// Check if we're in a GC task (mutation task is active)
	if c.mutation.taskState.name != "" {
		// GC is running, use snapshot
		protectedTS, isActive := c.getBackupProtectionSnapshot()
		if isActive {
			extraTS = protectedTS
		}
	} else {
		// Not in GC, use current protection state
		c.backupProtection.RLock()
		if c.backupProtection.isActive && time.Since(c.backupProtection.lastUpdateTime) <= 20*time.Minute {
			extraTS = c.backupProtection.protectedTS
		}
		c.backupProtection.RUnlock()
	}

	return c.mutation.snapshotMeta.GetSnapshot(c.ctx, c.sid, c.fs, c.mp, extraTS)
}

func (c *checkpointCleaner) GetSnapshotsLocked() (*logtail.SnapshotInfo, error) {
	// Use snapshot taken at GC start to ensure consistency
	var extraTS types.TS
	protectedTS, isActive := c.getBackupProtectionSnapshot()
	if isActive {
		extraTS = protectedTS
		logutil.Info(
			"GC-Backup-Protection-Add-Fake-Snapshot",
			zap.String("protected-ts", extraTS.ToString()),
		)
	}

	// Pass the protected TS to GetSnapshot, which will add it to cluster snapshots
	if !extraTS.IsEmpty() {
		return c.mutation.snapshotMeta.GetSnapshot(c.ctx, c.sid, c.fs, c.mp, extraTS)
	}
	return c.mutation.snapshotMeta.GetSnapshot(c.ctx, c.sid, c.fs, c.mp)
}
func (c *checkpointCleaner) GetTablePK(tid uint64) string {
	c.mutation.Lock()
	defer c.mutation.Unlock()
	return c.mutation.snapshotMeta.GetTablePK(tid)
}

func (c *checkpointCleaner) ISCPTables() (map[uint64]types.TS, error) {
	if c.iscpTablesFunc != nil {
		return c.iscpTablesFunc()
	}
	return c.mutation.snapshotMeta.GetISCP(c.ctx, c.sid, c.fs, c.mp)
}

func (c *checkpointCleaner) GetDetails(ctx context.Context) (map[uint32]*TableStats, error) {
	scan := c.GetScannedWindow()
	if scan == nil {
		return nil, nil
	}
	window := scan.Clone()
	return window.Details(ctx, c.mutation.snapshotMeta, c.mp)
}
